package version3.part2.Client.serviceCenter;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import version3.part2.Client.Cache.ServiceCache;
import version3.part2.Client.serviceCenter.ZKWatcher.watchZK;
import version3.part2.Client.serviceCenter.balance.LoadBalance;
import version3.part2.Client.serviceCenter.balance.impl.ConsistencyHashBalance;
import version3.part2.common.service.UserService;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;

public class ZKServiceCenter implements ServiceCenter {
    private CuratorFramework client;
    // zookeeper根节点
    private static final String ZK_REGISTER_ROOT_PATH="MyRpc";
    // 白名单
    private static final String RETRY_ROOT_PATH="CanRetry";
    private ServiceCache cache ;
    private Map<String, LoadBalance> loadBalanceMap;

    public ZKServiceCenter() throws InterruptedException {

        // 重试策略：初始重试时间为1s，重试3次,
        // 第1次重试间隔：1s,第2次重试间隔：2s,第3次重试间隔：4s
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        // 连接zookeeper
        this.client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:9090").sessionTimeoutMs(50000).
                retryPolicy(retryPolicy).namespace(ZK_REGISTER_ROOT_PATH).build();
        this.client.start();
        System.out.println("Connected to zookeeper");

        this.cache = new ServiceCache();

        watchZK watcher = new watchZK(client,cache);
        watcher.watchToUpdate("/"+ZK_REGISTER_ROOT_PATH);

        // 每个服务 都有一个负载均衡器
        loadBalanceMap = Map.of(
                UserService.class.getName(), new ConsistencyHashBalance()
        );

    }

    @Override
    public InetSocketAddress serviceDiscovery(String serviceName) {
        try{
            // 获取服务地址列表,先从缓存中获取
            List<String> serviceAddressList = cache.getService(serviceName);
            if(serviceAddressList == null){
                // 从zookeeper中获取服务地址列表,并缓存
                serviceAddressList = client.getChildren().forPath("/" + serviceName);
                // 如果zookeeper服务地址列表为空，则抛出异常
                if(serviceAddressList == null || serviceAddressList.size() == 0) {
                    throw new RuntimeException("未找到服务地址");
                }
            }
            // 负载均衡 获得服务地址
            String serviceAddress = loadBalanceMap.get(serviceName).balance(serviceAddressList);
            // 解析服务地址
            return parseAddress(serviceAddress);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public boolean isRetry(String serviceName) {
        boolean isRetry = false;
        try{
            List<String> serviceList = client.getChildren().forPath("/" + RETRY_ROOT_PATH);
            if(serviceList != null && serviceList.size() > 0){
                isRetry = serviceList.contains(serviceName);
            }
            if(isRetry) System.out.println("服务["+serviceName+"]在白名单上，可重试");
            else System.out.println("服务["+serviceName+"]不在白名单上，不可重试");
            return isRetry;
        }catch (Exception e){
            throw new RuntimeException(e);
        }
    }

    // String(ip:port) -> InetSocketAddress
    private InetSocketAddress parseAddress(String address) {
        String[] addressParts = address.split(":");
        return new InetSocketAddress(addressParts[0],Integer.parseInt(addressParts[1]));
    }

    // InetSocketAddress -> String(ip:port)
    private String getServiceAddress(InetSocketAddress address) {
        return address.getHostName() + ":" + address.getPort();
    }
}
